package com.atguigu.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.common.GmallConfig;
import com.atguigu.utils.DimUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.Collection;
import java.util.Set;

public class DimSinkFunction extends RichSinkFunction<JSONObject> {

    private Connection connection;

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    //value:{"database":"gmall","table":"base_trademark","type":"update","ts":1592270938,"xid":13090,"xoffset":1573,"data":{"id":"13","tm_name":"aaaa"},"old":null,"sinkTable":"dim_base_trademark"}
    @Override
    public void invoke(JSONObject value, Context context) throws Exception {

        //获取类型
        String type = value.getString("type");
        if ("bootstrap-start".equals(type) || "bootstrap-complete".equals(type)) {
            return;
        }

        //拼接SQL语句:upsert into db.tn(id,name,sex) values('..','..','..')
        String sinkTable = value.getString("sinkTable");
        JSONObject data = value.getJSONObject("data");
        String upsertSql = genSql(sinkTable, data);

        //打印SQL语句
        System.out.println(upsertSql);

        //编译SQL
        PreparedStatement preparedStatement = connection.prepareStatement(upsertSql);

        //判断,如果是更新操作,则需要将Redis中的维度数据清除
        if ("update".equals(value.getString("type"))) {
            DimUtil.deleteRedisDimInfo(sinkTable.toUpperCase(), data.getString("id"));
        }

        //执行写入操作并提交
        preparedStatement.execute();
        connection.commit();

        //释放资源
        preparedStatement.close();
    }

    //拼接SQL语句:upsert into db.tn(id,name,sex) values ('..','..','..')
    //{"id":"13","tm_name":"aaaa"}
    private String genSql(String sinkTable, JSONObject data) {

        //[id,name,sex] => "id,name,sex"
        Set<String> columns = data.keySet();
        //[aa,bbbb,ccc] => "'aa','bb','cc'"
        Collection<Object> values = data.values();

        return "upsert into " + GmallConfig.HBASE_SCHEMA + "." + sinkTable + "(" +
                StringUtils.join(columns, ",") + ") values ('" +
                StringUtils.join(values, "','") + "')";
    }
}
